package com.ebspp.client.serverManager;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;

import com.ebspp.global.EBSPPTimer;
import com.ebspp.global.General;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.Identity;
import com.ebspp.thrift.ResourceType;

/**
 * asynchronous garbage collection.
 * It accepts the garbage keys, stores them in buffer.
 * A thread clean up the buffer and request garbage collection asynchronously
 * 
 * @author x23pan
 *
 */
public class AsyncGBCollector implements IGarbageCollector
{
	private Identity id;
	private ResourceType resType;

	Map<String, LinkedBlockingQueue<DataMsg>> serverQueueMap;
	Map<String, LinkedBlockingQueue<DataMsg>> serverFailQueueMap;
	//async collect thread
	Collector collector;
	
	public AsyncGBCollector(Identity id, ResourceType type) throws TTransportException
	{
		this.id = id;	
		this.resType = type;
		serverQueueMap = Collections.synchronizedMap(new HashMap<String, LinkedBlockingQueue<DataMsg>>());
		serverFailQueueMap = Collections.synchronizedMap(new HashMap<String, LinkedBlockingQueue<DataMsg>>());
		
		Thread collectT = new Thread(collector);
		collectT.start();
	}


	/**
	 * called by the IPlacement
	 */
	@Override
	public boolean garbageCollect(DataMsg msg, String server) 
	{
		// TODO Auto-generated method stub
		if(serverQueueMap.get(server) == null)
		{
			GarbageConnection gbc = ServerManager.createGBConn(server, resType, id);
			if(gbc == null)
				return false;
			
			LinkedBlockingQueue<DataMsg> gbQueue = new LinkedBlockingQueue<DataMsg>();
			serverQueueMap.put(server, gbQueue);
		}
		serverQueueMap.get(server).add(msg);
		return true;
	}
	
	public List<DataMsg> getFailList(String server, int size)
	{
		if(size < 0)
			size = Integer.MAX_VALUE;
		
		if(serverFailQueueMap.get(server) == null)
			return null;
		else
		{
			List<DataMsg> failList = new ArrayList<DataMsg>();
			serverFailQueueMap.get(server).drainTo(failList, size);
			return failList;
		}
	}
	
	private boolean closeConnection(String server)
	{
		boolean result = ServerManager.closeGBConn(server, id);
		serverQueueMap.remove(server);
		serverFailQueueMap.remove(server);
		
		return result;
	}
	
	private void addFailQueue(List<DataMsg> failList, String server)
	{
		if(failList == null || failList.size() == 0)
			return;
		
		if(serverFailQueueMap.get(server) == null)
		{
			LinkedBlockingQueue<DataMsg> failQueue = new LinkedBlockingQueue<DataMsg>();
			serverFailQueueMap.put(server, failQueue);
		}
		
		LinkedBlockingQueue<DataMsg> failQueue = serverFailQueueMap.get(server);
		failQueue.addAll(failList);
		while(failQueue.size() > General.failGBSize)
			failQueue.remove();
	}
	
	/**
	 * currently the collector has the following tasks:
	 * 1.garbage collect from the remote servers.
	 * 2.close garbage connection if seldom used
	 * 
	 * @author x23pan
	 *
	 */
	class Collector implements Runnable
	{
		Map<String, Integer> unUsedCnt;
		
		public Collector()
		{
			unUsedCnt = new HashMap<String, Integer>();
		}
		
		@Override
		public void run() 
		{
			// TODO Auto-generated method stub
			while(true)
			{
				try {
					Thread.sleep(EBSPPTimer.garbageCollectT);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
				Map<String, List<DataMsg>> toCollect = new HashMap<String, List<DataMsg>>();
				
				Iterator<String> serverIt = serverQueueMap.keySet().iterator();
				synchronized(serverQueueMap)
				{
					while(serverIt.hasNext())
					{
						String server = serverIt.next();
						List<DataMsg> gbKeys = new ArrayList<DataMsg>();
						serverQueueMap.get(server).drainTo(gbKeys, Integer.MAX_VALUE);
						if(gbKeys.size() > 0)
							toCollect.put(server, gbKeys);
					}
				}
				
				for(String server : toCollect.keySet())
				{
					List<DataMsg> gbs = toCollect.get(server);
					try {
						List<DataMsg> failList = ServerManager.garbageCollect(gbs, server);
						if(failList.size() > 0)
							System.err.println("fail to collect garbage from: " + server);
						addFailQueue(failList, server);
					} 
					catch (TException e) {
						// TODO Auto-generated catch block
						System.err.println("garbage connection fail: " + gbs);
						e.printStackTrace();
					}
					
					unUsedCnt.put(server, 0);
				}
				
				serverIt = unUsedCnt.keySet().iterator();
				while(serverIt.hasNext())
				{
					String server = serverIt.next();
					if(toCollect.get(server) == null)
					{
						unUsedCnt.put(server, unUsedCnt.get(server)+1);
						if(unUsedCnt.get(server) > General.gbConnCloseCnt)
						{
							closeConnection(server);
							unUsedCnt.remove(server);
						}
					}
				}
	
			}
		}
	}
	
}

